package com.aotain.jupiter.flink;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import com.aotain.common.mongo.DataColumn;
import com.aotain.apollo.mongodb.MongoImportTask;
import com.aotain.jupiter.flink.MongoSinkObject.DML;

public class MongoDBSink implements SinkFunction<MongoSinkObject>{
	private DML dmlType = DML.INSERT;
	
	/**
	 * 
	 */
	private static final long serialVersionUID = -860615806507394645L;

	@Override
	public void invoke(MongoSinkObject value) throws Exception {
		// TODO Auto-generated method stub
		MongoImportTask task = new MongoImportTask();
		
		dmlType = value.getDmlType();
		
		if(dmlType == DML.INSERT) {
			HashMap<String,List<List<DataColumn>>> 
			importData = new HashMap<String,List<List<DataColumn>>>();
			List<List<DataColumn>> tableData = new ArrayList<List<DataColumn>>();
			tableData.add(value.getImportData());
			importData.put(value.getTableName(), tableData);
			task.InsertRowData(importData);
			task.taskCore();
		} else if (dmlType ==  DML.UPDATE) {
			HashMap<String,List<HashMap<List<DataColumn>,List<DataColumn>>>> tableMapUpdate 
				= new HashMap<String,List<HashMap<List<DataColumn>,List<DataColumn>>>>();
			
			List<HashMap<List<DataColumn>,List<DataColumn>>> 
				updateData = new ArrayList<HashMap<List<DataColumn>,List<DataColumn>>>();
			
			HashMap<List<DataColumn>,List<DataColumn>> 
				rowData = new HashMap<List<DataColumn>,List<DataColumn>>();
			rowData.put(value.getFilterData(), value.getImportData());
			
			tableMapUpdate.put(value.getTableName(), updateData);
			task.UpdateRowData(tableMapUpdate);
			task.taskCore();
		}
	}

}
